package com.atguigu.tingshu.order.delay;

import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.order.service.OrderInfoService;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author: atguigu
 * @create: 2023-12-02 11:56
 */
@Slf4j
@Component
public class DelayMsgConsumer {

    @Autowired
    private RedissonClient redissonClient;


    @Autowired
    private OrderInfoService orderInfoService;

    //项目启动后自动执行监听延迟队列消息方法
    @PostConstruct
    public void orderCanal() {
        //开启线程 采用线程池工具类
        log.info("开始监听延迟消息....");
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.submit(()->{
            RBlockingDeque<String> blockingQueue = redissonClient.getBlockingDeque(KafkaConstant.QUEUE_ORDER_CANCEL);
            while (true) {
                try {
                    String orderId = blockingQueue.take();
                    if (StringUtils.isNotBlank(orderId)) {
                        log.info("监听到延迟消息：{}", orderId);
                        orderInfoService.orderCancel(orderId);
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }

}
